from client import es_client
from client.es_client import ESResponseWrapper
from utils.logger_util import logger
import math


def _to_day(date_str: str) -> str:
    """提取 YYYY-MM-DD（假设 createDate 为 ISO 字符串）"""
    try:
        return date_str[:10]
    except Exception:
        return date_str


async def query_health_metrics(term: dict, start_time: str, end_time: str, imeis: list | None = None) -> dict:
    """
    通用健康指标查询：
    - 每天 08:00-18:00 的静息心率（dataType=心率，intensity<2）P20
    - 每天 00:00-06:00 的心率与血氧平均值

    参数：
      - term: 额外的 term 过滤（仿 card_dao 用法），例如 {"projectId": 123}
      - start_time / end_time: ISO 字符串（如 2025-11-01T00:00:00.000Z）
      - imeis: 可选，限制设备列表

    返回：
      {
        "resting_hr_P20_by_day": [{"date": "YYYY-MM-DD", "P20": float}],
        "night_avg_by_day": [{"date": "YYYY-MM-DD", "avg_hr": float, "avg_spo2": float}]
      }
    """

    es = await es_client.get_es_client()

    # 1) 08:00-18:00 静息心率 P20（心率 & intensity<2）
    must_clause = [
        {"term": {"dataType": "心率"}},
        {"range": {"createDate": {"gte": start_time, "lte": end_time}}},
        {"range": {"intensity": {"lt": 2}}},
        {
            "script": {
                "script": {
                    "source": "doc['createDate'].value.getHour() >= 8 && doc['createDate'].value.getHour() < 18",
                    "lang": "painless"
                }
            }
        }
    ]
    if term:
        must_clause.append({"term": term})
    if imeis:
        must_clause.append({"terms": {"imei": imeis}})

    query_hr_P20 = {
        "size": 0,
        "query": {"bool": {"must": must_clause}},
        "aggs": {
            "by_day": {
                "date_histogram": {
                    "field": "createDate",
                    "calendar_interval": "day",
                    "min_doc_count": 0
                },
                "aggs": {
                    "pcts": {"percentiles": {"field": "healthValue", "percents": [5]}}
                }
            }
        }
    }

    try:
        res_hr_P20 = await es.search(index="p_health_logs", query=query_hr_P20)
        aggs1 = ESResponseWrapper(res_hr_P20).aggs
        buckets1 = aggs1.get("by_day", {}).get("buckets", [])
        resting_hr_P20_by_day = []
        for b in buckets1:
            date = _to_day(b.get("key_as_string", ""))
            P20_val = None
            vals = b.get("pcts", {}).get("values") or {}
            # ES percentiles 返回键可能是字符串 "5.0"
            P20_val = vals.get("5.0") if "5.0" in vals else vals.get(5) or vals.get("5")
            resting_hr_P20_by_day.append({"date": date, "P20": P20_val})
    except Exception as e:
        logger.error(f"query resting_hr_P20_by_day failed: {e}")
        resting_hr_P20_by_day = []

    # 2) 00:00-06:00 心率与血氧平均值
    must_clause2 = [
        {"range": {"createDate": {"gte": start_time, "lte": end_time}}},
        {
            "script": {
                "script": {
                    "source": "doc['createDate'].value.getHour() >= 0 && doc['createDate'].value.getHour() < 6",
                    "lang": "painless"
                }
            }
        }
    ]
    if term:
        must_clause2.append({"term": term})
    if imeis:
        must_clause2.append({"terms": {"imei": imeis}})

    query_night_avg = {
        "size": 0,
        "query": {"bool": {"must": must_clause2}},
        "aggs": {
            "by_day": {
                "date_histogram": {
                    "field": "createDate",
                    "calendar_interval": "day",
                    "min_doc_count": 0
                },
                "aggs": {
                    "hr": {
                        "filter": {"term": {"dataType": "心率"}},
                        "aggs": {"avg_hr": {"avg": {"field": "healthValue"}}}
                    },
                    "spo2": {
                        "filter": {"term": {"dataType": "血氧"}},
                        "aggs": {"avg_spo2": {"avg": {"field": "healthValue"}}}
                    }
                }
            }
        }
    }

    try:
        res_night = await es.search(index="p_health_logs", query=query_night_avg)
        aggs2 = ESResponseWrapper(res_night).aggs
        buckets2 = aggs2.get("by_day", {}).get("buckets", [])
        night_avg_by_day = []
        for b in buckets2:
            date = _to_day(b.get("key_as_string", ""))
            avg_hr = b.get("hr", {}).get("avg_hr", {}).get("value")
            avg_spo2 = b.get("spo2", {}).get("avg_spo2", {}).get("value")
            night_avg_by_day.append({"date": date, "avg_hr": avg_hr, "avg_spo2": avg_spo2})
    except Exception as e:
        logger.error(f"query night_avg_by_day failed: {e}")
        night_avg_by_day = []

    return {
        "resting_hr_P20_by_day": resting_hr_P20_by_day,
        "night_avg_by_day": night_avg_by_day
    }


async def get_resting_hr_P20_by_day(term: dict, start_time: str, end_time: str, imeis: list | None = None):
    result = await query_health_metrics(term, start_time, end_time, imeis)
    return result.get("resting_hr_P20_by_day", [])


async def get_night_avg_hr_spo2_by_day(term: dict, start_time: str, end_time: str, imeis: list | None = None):
    result = await query_health_metrics(term, start_time, end_time, imeis)
    return result.get("night_avg_by_day", [])


def _compose_bool_query(query_bool: dict | None, must_additions: list, imeis: list | None):
    """组合 bool 查询，兼容传入的 query_bool，并追加必要的 must 条件。"""
    base_must = []
    base_must_not = []
    base_filter = []
    if query_bool:
        base_must = list(query_bool.get("must", []))
        base_must_not = list(query_bool.get("must_not", []))
        base_filter = list(query_bool.get("filter", []))
    base_must.extend(must_additions)
    if imeis:
        base_must.append({"terms": {"imei": imeis}})
    return {"bool": {"must": base_must, "must_not": base_must_not, "filter": base_filter}}


async def get_resting_hr_P20_by_imei_day(query_bool: dict | None, start_time: str, end_time: str,
                                         imeis: list | None = None):
    """
    每天 08:00-18:00 静息心率（dataType=心率, intensity<2）P20，按 IMEI + 天聚合。
    额外返回基于 P10 与 P90 的稳健标准差（近似）：std_p10_p90 ≈ (P90 - P10) / 2.563。
    返回：[{"imei": str, "date": "YYYY-MM-DD", "P20": float, "std_p10_p90": float | None}]
    """
    es = await es_client.get_es_client()
    must_additions = [
        {"term": {"dataType": "心率"}},
        {"range": {"createDate": {"gte": start_time, "lte": end_time}}}
    ]
    query = {
        "size": 0,
        "query": _compose_bool_query(query_bool, must_additions, imeis),
        "aggs": {
            "imei": {
                "terms": {"field": "imei", "size": 10000},
                "aggs": {
                    "day": {
                        "date_histogram": {
                            "field": "createDate",
                            "calendar_interval": "day",
                            "time_zone": "+08:00"
                        },
                        "aggs": {

                            "hr_all": {
                                "percentiles": {
                                    "field": "healthValue",
                                    "percents": [50]
                                }
                            },
                            "count_all": {
                                "value_count": {"field": "healthValue"}
                            },

                            "rest": {
                                "filter": {"range": {"intensity": {"lt": 2}}},
                                "aggs": {
                                    "hr_rest": {
                                        "percentiles": {"field": "healthValue", "percents": [50]}
                                    },
                                    "count_rest": {
                                        "value_count": {"field": "healthValue"}
                                    }
                                }
                            },

                            "active": {
                                "filter": {"range": {"intensity": {"gte": 2}}},
                                "aggs": {
                                    "hr_active": {
                                        "percentiles": {"field": "healthValue", "percents": [50]}
                                    },
                                    "count_active": {
                                        "value_count": {"field": "healthValue"}
                                    }
                                }
                            }
                        }
                    }
                }
            }
        }
    }
    try:
        res = await es.search(index="p_health_logs", query=query)
        return parse_es_daily_heart(res)

    except Exception as e:
        logger.error(f"get_resting_hr_P20_by_imei_day failed: {e}")
        return []


async def get_night_avg_hr_spo2_by_imei_day(query_bool: dict | None, start_time: str, end_time: str,
                                            imeis: list | None = None):
    """
    每天 00:00-06:00 的心率与血氧平均值，按 IMEI + 天聚合。
    额外返回当日（00:00-06:00）血氧值 <95 占非空样本的比例（spo2_ratio_lt95）。
    返回：[{"imei": str, "date": "YYYY-MM-DD", "avg_hr": float, "avg_spo2": float, "spo2_ratio_lt95": float | None}]
    """
    es = await es_client.get_es_client()
    must_additions = [
        {"range": {"createDate": {"gte": start_time, "lte": end_time}}},
        {
            "script": {
                "script": {
                    "source": "doc['createDate'].value.getHour() >= 0 && doc['createDate'].value.getHour() < 6",
                    "lang": "painless"
                }
            }
        }
    ]
    query = {
        "size": 0,
        "query": _compose_bool_query(query_bool, must_additions, imeis),
        "aggs": {
            "by_imei": {
                "terms": {"field": "imei", "size": 10000},
                "aggs": {
                    "by_day": {
                        "date_histogram": {
                            "field": "createDate",
                            "calendar_interval": "day",
                            "min_doc_count": 0
                        },
                        "aggs": {
                            "hr": {
                                "filter": {"term": {"dataType": "心率"}},
                                "aggs": {"avg_hr": {"avg": {"field": "healthValue"}}}
                            },
                            "spo2": {
                                "filter": {"term": {"dataType": "血氧"}},
                                "aggs": {
                                    "avg_spo2": {"avg": {"field": "healthValue"}},
                                    "vc_spo2": {"value_count": {"field": "healthValue"}},
                                    "lt95": {
                                        "filter": {"range": {"healthValue": {"lt": 95}}},
                                        "aggs": {"vc_lt95": {"value_count": {"field": "healthValue"}}}
                                    }
                                }
                            }
                        }
                    }
                }
            }
        }
    }
    try:
        res = await es.search(index="p_health_logs", query=query)
        aggs = ESResponseWrapper(res).aggs
        imei_buckets = aggs.get("by_imei", {}).get("buckets", [])
        rows = []
        for ib in imei_buckets:
            imei = ib.get("key")
            day_buckets = ib.get("by_day", {}).get("buckets", [])
            for db in day_buckets:
                date = _to_day(db.get("key_as_string", ""))
                avg_hr = db.get("hr", {}).get("avg_hr", {}).get("value")
                avg_spo2 = db.get("spo2", {}).get("avg_spo2", {}).get("value")
                tot = db.get("spo2", {}).get("vc_spo2", {}).get("value")
                lt = db.get("spo2", {}).get("lt95", {}).get("vc_lt95", {}).get("value")
                ratio_lt95 = None
                if isinstance(tot, (int, float)) and tot and isinstance(lt, (int, float)):
                    tmp = lt / tot if tot > 0 else None
                    ratio_lt95 = round(tmp, 4) if isinstance(tmp, (int, float)) and math.isfinite(tmp) else None
                rows.append(
                    {"imei": imei, "date": date, "avg_hr": avg_hr, "avg_spo2": avg_spo2, "spo2_ratio_lt95": ratio_lt95})
        return rows
    except Exception as e:
        logger.error(f"get_night_avg_hr_spo2_by_imei_day failed: {e}")
        return []


async def get_day_avg_hr_by_imei_day(query_bool: dict | None, start_time: str, end_time: str,
                                     imeis: list | None = None):
    """
    每天 06:00-22:00 的平均心率，按 IMEI + 天聚合。
    返回：[{"imei": str, "date": "YYYY-MM-DD", "avg_hr": float}]
    """
    es = await es_client.get_es_client()
    must_additions = [
        {"range": {"createDate": {"gte": start_time, "lte": end_time}}},
        {
            "script": {
                "script": {
                    "source": "doc['createDate'].value.getHour() >= 6 && doc['createDate'].value.getHour() < 22",
                    "lang": "painless"
                }
            }
        }
    ]
    query = {
        "size": 0,
        "query": _compose_bool_query(query_bool, must_additions, imeis),
        "aggs": {
            "by_imei": {
                "terms": {"field": "imei", "size": 10000},
                "aggs": {
                    "by_day": {
                        "date_histogram": {
                            "field": "createDate",
                            "calendar_interval": "day",
                            "min_doc_count": 0
                        },
                        "aggs": {
                            "hr": {
                                "filter": {"term": {"dataType": "心率"}},
                                "aggs": {"avg_hr": {"avg": {"field": "healthValue"}}}
                            }
                        }
                    }
                }
            }
        }
    }
    try:
        res = await es.search(index="p_health_logs", query=query)
        aggs = ESResponseWrapper(res).aggs
        imei_buckets = aggs.get("by_imei", {}).get("buckets", [])
        rows = []
        for ib in imei_buckets:
            imei = ib.get("key")
            day_buckets = ib.get("by_day", {}).get("buckets", [])
            for db in day_buckets:
                date = _to_day(db.get("key_as_string", ""))
                avg_hr = db.get("hr", {}).get("avg_hr", {}).get("value")
                rows.append({"imei": imei, "date": date, "avg_hr": avg_hr})
        return rows
    except Exception as e:
        logger.error(f"get_day_avg_hr_by_imei_day failed: {e}")
        return []


async def get_hr_timeseries_by_imei(query_bool: dict | None,
                                    start_time: str,
                                    end_time: str,
                                    imeis: list | None = None,
                                    size: int = 10000,
                                    from_offset: int | None = None) -> list:
    """
    拉取原始心率时序数据（按 createDate 升序），用于运动恢复心率检测。

    参数：
      - query_bool: 额外的 bool 查询条件（可包含 filter/must_not 等）
      - start_time/end_time: 时间范围（ES 可解析的时间字符串）
      - imeis: 可选，限定设备列表（terms 过滤）
      - size: 返回最大文档数，默认 10000

    返回：每条记录一个字典，字段包含：
      {"imei": str, "createDate": str, "hr": float | None, "intensity": float | None}
    """
    es = await es_client.get_es_client()
    start_time = start_time.replace('+00:00', '.000Z')
    end_time = end_time.replace('+00:00', '.000Z')
    must_additions = [
        {"term": {"dataType": "心率"}},
        {"range": {"createDate": {"gte": start_time, "lte": end_time}}}
    ]
    query = {
        "from": from_offset or 0,
        "size": size,
        "query": _compose_bool_query(query_bool, must_additions, imeis),
        "sort": [
            {"createDate": {"order": "asc"}},
            {"imei": {"order": "asc"}}
        ],
        "_source": ["imei", "createDate", "healthValue", "intensity", "dataType"]
    }
    try:
        res = await es.search(index="p_health_logs", query=query)
        wrapper = ESResponseWrapper(res)
        rows = []
        for doc in wrapper.docs:
            # 转数值，避免字符串干扰后续计算
            hv = doc.get("healthValue")
            try:
                hr = float(hv) if hv is not None else None
            except Exception:
                hr = None
            it = doc.get("intensity")
            try:
                intensity = float(it) if it is not None else None
            except Exception:
                intensity = None
            rows.append({
                "imei": doc.get("imei"),
                "createDate": doc.get("createDate"),
                "hr": hr,
                "intensity": intensity
            })
        return rows
    except Exception as e:
        logger.error(f"get_hr_timeseries_by_imei failed: {e}")
        return []


async def get_hr_timeseries_by_imei_page(query_bool: dict | None,
                                         start_time: str,
                                         end_time: str,
                                         imeis: list | None = None,
                                         size: int = 5000,
                                         search_after: list | None = None) -> dict:
    """
    使用 search_after 深分页拉取心率时序页面，返回 rows 与下一页的 search_after 标记。
    避免触发 from+size 的 result_window 限制。
    """
    es = await es_client.get_es_client()
    start_time = start_time.replace('+00:00', '.000Z')
    end_time = end_time.replace('+00:00', '.000Z')
    must_additions = [
        {"term": {"dataType": "心率"}},
        {"range": {"createDate": {"gte": start_time, "lte": end_time}}}
    ]
    query: dict = {
        "size": size,
        "query": _compose_bool_query(query_bool, must_additions, imeis),
        "sort": [
            {"createDate": {"order": "asc"}},
            {"imei": {"order": "asc"}}
        ],
        "_source": ["imei", "createDate", "healthValue", "intensity", "dataType"]
    }
    if search_after:
        query["search_after"] = search_after

    try:
        res = await es.search(index="p_health_logs", query=query)
        wrapper = ESResponseWrapper(res)
        rows = []
        for hit in wrapper.hits:
            src = hit.get("_source", {})
            hv = src.get("healthValue")
            try:
                hr = float(hv) if hv is not None else None
            except Exception:
                hr = None
            it = src.get("intensity")
            try:
                intensity = float(it) if it is not None else None
            except Exception:
                intensity = None
            rows.append({
                "imei": src.get("imei"),
                "createDate": src.get("createDate"),
                "hr": hr,
                "intensity": intensity
            })
        last_sort = wrapper.hits[-1].get("sort") if wrapper.hits else None
        return {"rows": rows, "last_sort": last_sort}
    except Exception as e:
        logger.error(f"get_hr_timeseries_by_imei_page failed: {e}")
        return {"rows": [], "last_sort": None}


def parse_es_daily_heart(res):
    rows = []
    for imei_bucket in res["aggregations"]["imei"]["buckets"]:
        imei = imei_bucket["key"]
        for day_bucket in imei_bucket["day"]["buckets"]:
            date = day_bucket["key_as_string"][:10]

            hr_all = day_bucket["hr_all"]["values"].get("50.0")

            hr_rest = day_bucket["rest"]["hr_rest"]["values"].get("50.0")

            hr_active = day_bucket["active"]["hr_active"]["values"].get("50.0")
            count_all = day_bucket["count_all"]["value"]

            count_rest = day_bucket["rest"]["count_rest"]["value"]

            count_active = day_bucket["active"]["count_active"]["value"]

            rows.append({
                "imei": imei,
                "date": date,
                "hr_all_median": hr_all,
                "hr_rest_median": hr_rest,
                "hr_active_median": hr_active,
                "count_all": count_all,
                "count_rest": count_rest,
                "count_active": count_active,
            })

    return rows
